package cn.aiweiyi.qingjing.coupon.distribution.service.impl;

import cn.aiweiyi.qingjing.coupon.comm.constant.Constant;
import cn.aiweiyi.qingjing.coupon.comm.constant.CouponStatus;
import cn.aiweiyi.qingjing.coupon.comm.exception.CouponException;
import cn.aiweiyi.qingjing.coupon.comm.vo.AcquireCouponRequest;
import cn.aiweiyi.qingjing.coupon.comm.vo.CouponTemplateSDK;
import cn.aiweiyi.qingjing.coupon.comm.vo.GoodsInfo;
import cn.aiweiyi.qingjing.coupon.comm.vo.SettlementInfo;
import cn.aiweiyi.qingjing.coupon.distribution.dao.CouponDao;
import cn.aiweiyi.qingjing.coupon.distribution.entity.Coupon;
import cn.aiweiyi.qingjing.coupon.distribution.feign.SettlementClient;
import cn.aiweiyi.qingjing.coupon.distribution.feign.TemplateClient;
import cn.aiweiyi.qingjing.coupon.distribution.service.RedisService;
import cn.aiweiyi.qingjing.coupon.distribution.service.UserCouponService;
import cn.aiweiyi.qingjing.coupon.distribution.vo.CouponClassify;
import cn.aiweiyi.qingjing.coupon.distribution.vo.CouponKafkaMessage;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author 青衿
 * @Date 2020/11/21 19:44
 * 用户优惠券实现类,所有优惠券操作都记录在Redis,在通过Kafka同步到MySql中
 * 为什么用Kafka而不用你Spring Boot中的异步处理?因为以Spring Boot中的
 * 异步任务方式执行,如果失败了,就很难重新执行或恢复,而异步消息放到Kafka中,即使
 * 失败了,也很容易回溯记录,并重写获取消息进行执行,确保Redis和MySqk的一致性
 **/
@Slf4j
@Service
public class UserCouponServiceImpl implements UserCouponService {

    /**
     * Coupon Dao
     */
    @Autowired
    private CouponDao couponDao;

    /**
     * Redis服务
     */
    @Autowired
    private RedisService redisService;

    /**
     * 模板微服务客户端
     */
    @Autowired
    private TemplateClient templateClient;

    /**
     * 结算微服务客户端
     */
    @Autowired
    private SettlementClient settlementClient;

    /**
     * Kafka客户端
     */
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 根据状态查询用户的优惠券,该方法执行的逻辑是:
     * 1.先从Redis查询优惠券,如果为空,就从数据库查询,在存入Redis
     * 也为空就直接返回,否则向数据库查到的优惠券信息填充模板数据后,在存入Redis
     * 2.判断如果查询的状态是USABLE状态的,就对优惠券集合按三种状态进行分类,把已过期的
     * 优惠券添加到用户USABLE缓存集合中,并向Kafka队列发布一个某优惠券已过期的信息.
     * Kafka队列的消费端收到该消息,会在MySQL中执行Update操作,操作优惠券状态
     *
     * @param userId 用户id
     * @param status 优惠券状态
     * @return 优惠券集合
     * @throws CouponException
     */
    @Override
    public List<Coupon> findCouponsByStatus(Long userId, Integer status) throws CouponException {
        List<Coupon> cachedCoupons = redisService.getCachedCoupons(userId, status);

        List<Coupon> preTarget = null;
        if (CollectionUtils.isNotEmpty(cachedCoupons)) {
            log.debug("用户:{}的状态:{}优惠券缓存不为空", userId, status);
            preTarget = cachedCoupons;
        } else {
            log.debug("用户:{}的状态:{}优惠券缓存为空,从数据库查询", userId,
                    status);

            List<Coupon> dbCoupons = couponDao.findAllByUserIdAndStatus(
                    userId, CouponStatus.of(status));

            if (CollectionUtils.isEmpty(dbCoupons)) {
                log.debug("用户:{}在数据库中没有{}状态的优惠券", userId, status);
                return dbCoupons;
            }

            List<Integer> dbCouponIds = dbCoupons
                    .stream().map(Coupon::getTemplateId)
                    .collect(Collectors.toList());

            Map<Integer, CouponTemplateSDK> id2TemplateSDK =
                    templateClient.findIds2TemplateSDK(dbCouponIds)
                            .getData();

            dbCoupons.forEach(dc ->
                    dc.setTemplateSDK(id2TemplateSDK.get(dc.getTemplateId())));

            preTarget = dbCoupons;

            redisService.addCouponToCache(userId, preTarget, status);
        }

        //将缓存中优惠券id为-1的剔除掉(如果有)
        preTarget = preTarget.stream().filter(c -> c.getId() != -1)
                .collect(Collectors.toList());

        if (CouponStatus.of(status) == CouponStatus.USABLE) {
            CouponClassify classify = CouponClassify.classify(preTarget);

            if (CollectionUtils.isNotEmpty(classify.getExpired())) {
                //把每个过期的加入到用户过期优惠券的缓存
                redisService.addCouponToCache(
                        userId, classify.getExpired(),
                        CouponStatus.EXPIRED.getCode()
                );

                //收集所有故去优惠券的id
                List<Integer> expiredIds = classify.getExpired().stream()
                        .map(Coupon::getId).collect(Collectors.toList());

                //向kafka消息队列发过期id
                kafkaTemplate.send(Constant.TOPIC,
                        JSON.toJSONString(new CouponKafkaMessage(
                                CouponStatus.EXPIRED.getCode(),
                                expiredIds
                        )));
            }
            return classify.getUsable();
        }
        return null;
    }

    /**
     * 根据id查找用户领取的优惠券模板
     *
     * @param userId 用户id
     * @return 可领取的优惠券模板集合
     * @throws CouponException
     */
    @Override
    public List<CouponTemplateSDK> findAvailableTemplate(Long userId) throws CouponException {
        long curTime = new Date().getTime();
        List<CouponTemplateSDK> templateSDKS =
                templateClient.findAllUsableTemplate().getData();

        log.debug("查找到的所有模板(From TemplateClient)数量:{}",
                templateSDKS.size());

        //过滤过期的优惠券模板
        templateSDKS = templateSDKS.stream().filter(
                t -> t.getRule().getExpiration().getDeadline() > curTime
        ).collect(Collectors.toList());

        log.info("可用模板的数量:{}", templateSDKS.size());

        //key是模板id
        //value中的key是模板领取数量限制,value是优惠券模板完整信息
        Map<Integer, Pair<Integer, CouponTemplateSDK>> limit2Template =
                new HashMap<>(templateSDKS.size());

        //循环把
        templateSDKS.forEach(
                t -> limit2Template.put(
                        t.getId(),
                        Pair.of(t.getRule().getLimit(), t)
                )
        );
        List<CouponTemplateSDK> result =
                new ArrayList<>(limit2Template.size());

        //查询用户可用优惠券
        List<Coupon> userUsableCoupons = findCouponsByStatus(
                userId, CouponStatus.USABLE.getCode()
        );

        log.debug("Current User Has Usable Coupons:{},{}", userId,
                userUsableCoupons.size());

        //对用户可用优惠券集合按模板分组,得到的Map key是模板id
        Map<Integer, List<Coupon>> templateId2Coupons = userUsableCoupons
                .stream()
                .collect(Collectors.groupingBy(Coupon::getTemplateId));

        //根据Template的Rule判断是否可以领取优惠券模板
        limit2Template.forEach((k, v) -> {
            int limit = v.getLeft();
            CouponTemplateSDK templateSDK = v.getRight();
            if (templateId2Coupons.containsKey(k)
                    && templateId2Coupons.get(k).size() >= limit) {
                return;
            }
            result.add(templateSDK);
        });
        return result;
    }

    /**
     * 用户领取优惠券
     * 1.从TemplateCLient拿到对应的优惠券,并检查是否过期
     * 2.根据limitation判断用户是否可以领取
     * 3.保存到数据库
     * 4.填充COuponTemplateSDK
     * 5.保存到缓存
     *
     * @param request 凤凰组昂了模板信息的请求对象 {@link AcquireCouponRequest}
     * @return 领到的优惠券
     * @throws CouponException
     */
    @Override
    public Coupon acquireCoupon(AcquireCouponRequest request) throws CouponException {
        System.out.println("進入了");
        Map<Integer, CouponTemplateSDK> id2Template =
                templateClient.findIds2TemplateSDK(
                        Collections.singletonList(
                                request.getTemplateSDK().getId()))
                        .getData();

        //优惠券模板是需要存在的
        if (id2Template.size() <= 0) {
            log.error("TemplateClient无法获得模板:{}",
                    request.getTemplateSDK().getId());
            throw new CouponException("TemplateClient无法获得模板");
        }

        //判断用户是否领取该模板的券,以及是否超出领取限制
        List<Coupon> userUsableCoupons = findCouponsByStatus(
                request.getUserId(), CouponStatus.USABLE.getCode()
        );

        Map<Integer, List<Coupon>> templateId2Coupons = userUsableCoupons
                .stream()
                .collect(Collectors.groupingBy(Coupon::getTemplateId));

        if (templateId2Coupons.containsKey(request.getTemplateSDK().getId())
                && templateId2Coupons.get(request.getTemplateSDK().getId()).size() >=
                request.getTemplateSDK().getRule().getLimit()) {
            log.error("超出模板的领取数据限制:{}",
                    request.getTemplateSDK().getId());
            throw new CouponException("超出模板的领取数量限制");
        }

        //尝试去获取优惠券码
        String couponCode = redisService.tryToAcquireCouponCodeFromCache(
                request.getTemplateSDK().getId()
        );

        if (StringUtils.isEmpty(couponCode)) {
            log.error("无法获取优惠券码{}",
                    request.getTemplateSDK().getId());
            throw new CouponException("无法获取优惠券码");
        }

        Coupon newCoupon = new Coupon(
                request.getTemplateSDK().getId(), request.getUserId(),
                couponCode, CouponStatus.USABLE
        );

        newCoupon = couponDao.save(newCoupon);
        //填充Coupon对象的CouponTemplateSDK,一定会要放入缓存之前去填充
        newCoupon.setTemplateSDK(request.getTemplateSDK());
        //领取成功,把该优惠券放入用户缓存的USABLE集合中
        redisService.addCouponToCache(
                request.getUserId(),
                Collections.singletonList(newCoupon),
                CouponStatus.USABLE.getCode()
        );
        return newCoupon;
    }

    /**
     * 用户领取优惠券
     * 1.判断如果没有优惠券和模板信息,就直接训话你累加商品金额
     * 2.校验要结算的优惠券是否是用户自己的
     * 3.调用结算微服务进行结算
     * 4.判断如果是核销
     * 5.填充COuponTemplateSDK
     * 6.保存到缓存
     * 7.保存到数据库
     *
     * @param info 结算信息{@link SettlementInfo}
     * @return 结算后的信息(商品 、 券和金额)
     * @throws CouponException
     */
    @Override
    public SettlementInfo settlement(SettlementInfo info) throws CouponException {
        //【当没有传递优惠券时,直接返回商品总价】
        List<SettlementInfo.CouponAndTemplateInfo> ctInfos =
                info.getCouponAndTemplateInfos();
        if (CollectionUtils.isEmpty(ctInfos)) {
            log.info("本次结算不携带任何优惠券.");
            double goodsSum = 0.0;

            for (GoodsInfo gi : info.getGoodsInfos()) {
                goodsSum += gi.getPrice() + gi.getCount();//直接累加单价
            }

            //没有优惠券也就不存在优惠券的核销,SettlementInfo其他的字段不需要修改
            info.setCost(retain2Decimals(goodsSum));
        }

        //【校验传递的优惠券是否是自己的】
        List<Coupon> coupons = findCouponsByStatus(
                info.getUserId(), CouponStatus.USABLE.getCode()
        );

        Map<Integer, Coupon> id2Coupon = coupons.stream()//优惠券List转Map
                .collect(Collectors.toMap(
                        Coupon::getId,
                        Function.identity()
                ));
        //判断要结算的券id集合,是否是可用券id集合的子集,不是说明数据有问题不能结算
        if (MapUtils.isEmpty(id2Coupon) ||
                !CollectionUtils.isSubCollection(ctInfos.stream().map(
                        SettlementInfo.CouponAndTemplateInfo::getId)
                        .collect(Collectors.toList()), id2Coupon.keySet())) {
            log.info("{}", id2Coupon.keySet());
            log.info("{}", ctInfos.stream()
                    .map(SettlementInfo.CouponAndTemplateInfo::getId)
                    .collect(Collectors.toList()));
            log.error("要结算的优惠券数据有问题,不是该用户可用优惠券的子集");
            throw new CouponException("要结算的优惠券数据有问题,不是该用户可用优惠券的子集!");
        }
        log.debug("为用户结算优惠券数量:{}", ctInfos.size());

        List<Coupon> settleCoupons = new ArrayList<>(ctInfos.size());
        ctInfos.forEach(ci -> settleCoupons.add(id2Coupon.get(ci.getId())));

        //调用结算微服务获取结算信息,返回processedInfo包含结算结果
        SettlementInfo processedInfo =
                settlementClient.couputeRule(info).getData();

        //如果需要核销,且处理结果中优惠券和模板信息不为空
        if (processedInfo.getEmploy() && CollectionUtils.isNotEmpty(
                processedInfo.getCouponAndTemplateInfos()
        )) {
            log.info("为用户:{}核销优惠券:{}", info.getUserId(),
                    JSON.toJSONString(settleCoupons));
            //核销后,优惠券将更新到用户缓存的USED集合中(也意味着从USABLE集合移除)
            redisService.addCouponToCache(
                    info.getUserId(),
                    settleCoupons,
                    CouponStatus.USED.getCode()
            );

            //更新数据库
            kafkaTemplate.send(
                    Constant.TOPIC,
                    JSON.toJSONString(new CouponKafkaMessage(
                            CouponStatus.USED.getCode(),
                            settleCoupons.stream().map(Coupon::getId)
                                    .collect(Collectors.toList())
                    ))
            );
        }
        return processedInfo;
    }

    /**
     * 保留两位小数
     *
     * @param value
     * @return
     */
    private Double retain2Decimals(double value) {
        //BigDecimal.ROUND_HALF_UP 代表四舍五入
        return new BigDecimal(value)
                .setScale(2, BigDecimal.ROUND_HALF_UP)
                .doubleValue();
    }
}
